
#include "as_connector.h"


#ifdef _WIN32
LPFN_CONNECTEX as::inet::AsConnector::m_connect_fun = nullptr;
#endif

std::map<int, std::function<void()> > as::inet::AsConnector::m_signal_map;
namespace as
{
namespace inet
{
AsConnector::AsConnector(/* args */)
{
#ifdef _WIN32
    WSADATA data;
    WORD wVersion = MAKEWORD(2, 2);
    int err = WSAStartup(wVersion, &data);
    assert(err == 0);
#endif
    m_connectSocket = std::make_shared<as::inet::AsSession>();
#ifdef _WIN32
    m_ioPort = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0);
#else
    m_ioPort = epoll_create1(EPOLL_CLOEXEC);
#endif   

}

AsConnector::~AsConnector()
{
}

AsConnector* AsConnector::set_workthread_num(int num)
{
    m_workThreadNum = num;
    return this;
}

AsConnector* AsConnector::start(NetAddr &addr)
{
    m_addr = addr;
    m_connectSocket->init();
    if(m_connectSocket->connect(addr))
    {
        m_state = io_state::RUNNING;
    }
    else
    {
        m_state = io_state::STOP;
        return this;
    }

    register_sys_fun();
    
    auto ioContext = m_connectSocket->pop_io_context();
    ioContext->type = as::inet::IOType::Read;
    ioContext->socket = m_connectSocket->get_socket_fd(); 
    ioContext->ioSession = m_connectSocket;
#ifdef _WIN32
    HANDLE h = CreateIoCompletionPort((HANDLE)m_connectSocket->get_socket_fd(), m_ioPort, NULL, 0);
    //post recv event

    DWORD rLenght = 0;
    DWORD flags = 0;
    int ret = WSARecv(m_connectSocket->get_socket_fd(), &ioContext->wsaBuf, 1, &rLenght, &flags, &ioContext->overlapped, 0);
    if(ret == SOCKET_ERR_FD)
    {
        int err = AsSocket::get_last_err();
        if(err != WSA_IO_PENDING)
        {
            m_connectSocket->push_io_context(ioContext);
            m_connectSocket->close();
            std::cout << "connector start (host, port) error post recv:" << err << std::endl;
            return this;
        }
    }

#else
    
#endif
    if(m_new_con_cb)
    {
        m_new_con_cb(ioContext->ioSession);
    }

    int addThreadNum = m_workThreadNum - (int)m_workThreads.size();
    if(addThreadNum > 0)
    {
        for(int i = 0; i < addThreadNum; ++i)
        {
            m_workThreads.emplace_back(std::thread(&AsConnector::work_run, this));
        }
    }

    return this;
}

AsConnector* AsConnector::start(std::string_view ip)
{
    return this;
}

AsConnector* AsConnector::start(std::string_view host, int port)
{
    m_addr.set_addr(host, port);
    if(!m_connectSocket->init())
    {
        m_state = io_state::STOP;
        return this;
    }

    if(m_connectSocket->connect(m_addr))
    {
        m_state = io_state::RUNNING;
    }
    else
    {
        m_state = io_state::STOP;
        return this;
    }

    register_sys_fun();

    //post recv event
    auto ioContext = m_connectSocket->pop_io_context();
    ioContext->type = as::inet::IOType::Read;
    ioContext->socket = m_connectSocket->get_socket_fd();
    ioContext->ioSession = m_connectSocket;
#ifdef _WIN32

    HANDLE h = CreateIoCompletionPort((HANDLE)m_connectSocket->get_socket_fd(), m_ioPort, (ULONG_PTR)&ioContext->overlapped, 0);

    DWORD rLenght = 0;
    DWORD flags = 0;
    // printf("start ptr:%x %x\n", &ioContext, &ioContext->overlapped);
    int ret = WSARecv(m_connectSocket->get_socket_fd(), &ioContext->wsaBuf, 1, &rLenght, &flags, &ioContext->overlapped, 0);
    if(ret == SOCKET_ERR_FD)
    {
        int err = AsSocket::get_last_err();
        if(err != WSA_IO_PENDING)
        {
            m_connectSocket->push_io_context(ioContext);
            m_connectSocket->close();
            std::cout << "connector start (host, port) error post recv:" << err << std::endl;
            return this;
        }
    }
#else
    
    
#endif
    if(m_new_con_cb)
    {
        m_new_con_cb(ioContext->ioSession);
    }

    int addThreadNum = m_workThreadNum - (int)m_workThreads.size();
    if(addThreadNum > 0)
    {
        for(int i = 0; i < addThreadNum; ++i)
        {
            m_workThreads.emplace_back(std::thread(&AsConnector::work_run, this));
        }
    }

    return this;
}

AsConnector* AsConnector::async_connect(std::string_view host, int port)
{
    m_state = io_state::RUNNING;
    m_addr.set_addr(host, port);

    m_connectSocket->init();
    register_sys_fun();

    socket_fd fd = m_connectSocket->get_socket_fd();
    auto ret = SOCKET_ERR_FD;

    sockaddr_in localAddr = {0};
    localAddr.sin_family = AF_INET;
    localAddr.sin_addr.s_addr = INADDR_ANY;
    localAddr.sin_port = 0;
    ret = ::bind(fd, (const sockaddr*)&localAddr, sizeof(localAddr)); 
    if(ret == SOCKET_ERR_FD)
    {
        std::cout << "connector async connect bind error:" << AsSocket::get_last_err() << std::endl;
        return this;
    }

    AsSocket::set_keep_alive(fd);
#ifdef _WIN32
    if(CreateIoCompletionPort((HANDLE)fd, m_ioPort, NULL, 0) != m_ioPort)
    {
        m_connectSocket->close();
        assert(false);
        return this;
    }

    auto ioContext = m_connectSocket->pop_io_context();
    ioContext->type = as::inet::IOType::Connect;
    ioContext->socket = fd; 

    auto &addripv4 = m_addr.getIpv4();
connectlaber:
    ret = m_connect_fun(fd, (const sockaddr*)&addripv4, m_addr.get_len(), nullptr, 0, nullptr, &ioContext->overlapped);
    if(ret == false)
    {
        int err = AsSocket::get_last_err();
        if(err != WSA_IO_PENDING)
        {
            if(err == WSAECONNREFUSED || err == WSAENETUNREACH || err == WSAETIMEDOUT)
            {
                as::tool::sleepms(500);
                goto connectlaber;
            }
            m_connectSocket->push_io_context(ioContext);
            m_connectSocket->close();
            std::cout << "connector start (host, port) error post recv:" << err << std::endl;
            return this;
        }
    }
#else

#endif

    for(int i = 0; i < m_workThreadNum; ++i)
    {
        m_workThreads.emplace_back(std::thread(&AsConnector::work_run, this));
    }

    return this;
}

AsConnector* AsConnector::reconnect()
{
    return start(m_addr);
}

void AsConnector::register_sys_fun()
{
#ifdef _WIN32
    GUID guidConectEx = WSAID_CONNECTEX;
    DWORD dwBytes = 0;
    assert(SOCKET_ERROR != WSAIoctl(
        m_connectSocket->get_socket_fd(),
        SIO_GET_EXTENSION_FUNCTION_POINTER,
        &guidConectEx,
        sizeof(guidConectEx),
        &m_connect_fun,
        sizeof(m_connect_fun),
        &dwBytes,
        nullptr,
        nullptr
    ));
#endif
}

void AsConnector::create_io_and_post_event()
{
#ifdef _WIN32
    for(auto &io : m_events)
    {
        if(post_connect(&io) == false)
        {
            io.clear();
        }
    }

#endif
}

void AsConnector::stop()
{
    m_state = io_state::STOP;
    
    m_connectSocket->close();
    for(auto &thread : m_workThreads)
    {
        //auto &thread = m_workThreads[i];
        while(thread.joinable() == false)
        {
            std::this_thread::sleep_for(std::chrono::milliseconds(300));
        }

        thread.join();
    }

    m_state = io_state::STOP_OK;

}

bool AsConnector::post_connect(IOContext *io)
{
    auto err = AsSocket::get_last_err();
    std::cout << "post connect err:" << err << std::endl;
    if(err == 0)
    {
        // io->type = as::inet::IOType::Read;
        // PostQueuedCompletionStatus(m_ioPort, 0, NULL, &io->overlapped);
        // return true;
        return post_recv(io);
    } 

    return false;
}

bool AsConnector::post_recv(IOContext *io)
{
    io->clear();
    io->type = IOType::Read;
#ifdef _WIN32
    DWORD out_num = 0;
    DWORD out_flags = 0; 
    int err = WSARecv(io->socket, &io->wsaBuf, 1, &out_num, &out_flags, 
        &io->overlapped, nullptr);
    if(err == SOCKET_ERR_FD)
    {
        err = AsSocket::get_last_err();
        if(err != ERROR_IO_PENDING)
        {
            std::cout << "post recv error:" << err << std::endl;
            return false;
        }
    }
#else
#endif
    return true;
}

int AsConnector::DoConnect(IOContext *io)
{
#ifdef _WIN32
    if(setsockopt(io->socket, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, nullptr, 1) == false)
    {
        int errcode = AsSocket::get_last_err();
        std::cout << "do connect:" << errcode << std::endl; 
        return -1;
    }
#endif

    post_recv(io); 
    return 0;
}

int AsConnector::DoRecv(IOContext *io, int byteNum)
{
    if(m_recv_cb)
    {
        
    }
    io->clear();
    if(post_recv(io) == false)
    {
        AsSocket::close(io->socket);
        return -1;
    }
    return 0;
}

int AsConnector::DoWrite(IOContext *io, int byteNum)
{
    return 0;
}

int AsConnector::DoClose(IOContext *io)
{
    if(m_close_cb)
    {
        m_close_cb(io->ioSession);
    }

    io->ioSession->close();
    m_state = io_state::STOP;
    return 0;
}

void AsConnector::work_run()
{
#ifdef _WIN32
    OVERLAPPED_ENTRY overlappeds[SVC_MAX_EVENT_NUM]; 
#endif
    while (m_state != io_state::STOP_OK)
    {
#ifdef _WIN32
        ULONG eventNum = 0;
        if(!GetQueuedCompletionStatusEx(m_ioPort, &overlappeds[0], SVC_MAX_EVENT_NUM, &eventNum, INFINITE, false))
        {
            auto err = AsSocket::get_last_err();
            if(WAIT_TIMEOUT != err)
            {
                //LOG_INFO << "error" << std::endl;
            }

            as::tool::sleepms(300);
            continue;
        }

        if(m_state != io_state::RUNNING)
        {
            break;
        }

        if(eventNum > 0)
        {
            for(int i = 0; i < (int)eventNum; ++i)
            {
                auto &event = overlappeds[i];
				LPOVERLAPPED &ol = event.lpOverlapped;
				if (ol == 0) {
					std::cout << ("[iocp]GetQueuedCompletionStatusEx, no packet dequeue");
					return;
				}

                DWORD transNum = event.dwNumberOfBytesTransferred;
                IOContext *io= (CONTAINING_RECORD(ol, IOContext, overlapped));
                //printf("work run:%x %x\n", io, ol);
                if(io)
                {
                    std::cout << "idx:" << i << " type:" << io->type << " num:" << transNum << std::endl;
                    switch(io->type)
                    {
                        case IOType::Connect:
                        {
                            post_connect(io);
                            as::tool::sleepms(300);
                            break;
                        }
                        case IOType::Read:
                        {
                            if (transNum <= 0)
                            {
                                DoClose(io);
                                break;
                            }
                            DoRecv(io, (int)transNum);
                            break;
                        }
                        case IOType::Write:
                        {
                            if(transNum <= 0)
                            {
                                DoClose(io);
                                break;
                            }
                            DoWrite(io, transNum);
                            break;
                        }
                    }
                }
            }
        }
#else
        int n = ::epoll_wait(m_ioPort, &m_events[0], SVC_MAX_EVENT_NUM, -1); 
        if(n < 0)
        {
            LOG_ERROR << "epoll wait empty event:" << n <<  std::endl;
            return;
        }

        for(int i = 0; i < n; ++i)
        {
            IOContext *io = (IOContext*)m_events[i].data.ptr;
            if(io)
            {
                // LOG_INFO << "event fd:" << io->socket << " listen socket fd:" << m_listenSocket->get_socket_fd() << std::endl;
                if(io->socket == m_connectSocket->get_socket_fd())
                {
                    // DoAccept(io);
                }
                else
                {
                    auto clientSession = io->ioSession;
                    if(clientSession)
                    {
                        LOG_INFO << "io type:" << io->type << " enum:" << IOType::Read << std::endl;
                        if(io->type == IOType::Read)
                        {
                            int curPos = 0;
                            int ret = 0;
                            auto tmp = &io->buffer[0];
                            do
                            {
                                ret = AsSocket::recv(clientSession->get_socket_fd(), tmp, IO_CONTEXT_BUF_SIZE);
                                if(ret > 0)
                                {
                                    curPos += ret;
                                    tmp = &io->buffer[curPos];
                                    LOG_INFO << "recv again:" << ret << std::endl;
                                }
                                else if (ret < 0)
                                {
                                    LOG_INFO << "recv wait ..." << std::endl;
                                    SLEEP_MS(20);
                                }
                            } while (ret == 0);

                            if(m_recv_cb)
                            {
                                m_recv_cb(io->ioSession, io->buffer, curPos);
                            }
                        }
                    }
                }
            }     
        }
#endif
    }


    //LOG_INFO << "work thread break:" << std::this_thread::get_id() << std::endl;
}

void AsConnector::run()
{
    while(m_state != io_state::STOP_OK)
    {

    }
}









}
}